package org.hyf.inspur.LessonDesin.clear.count7;

import org.hyf.inspur.LessonDesin.clear.tools.TProperties;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public  class UserPortraitMapper extends Mapper<LongWritable, Text, Text, Text> {
    private Text okey = new Text();
    private Text ovalue = new Text();
    // 分类关系地址库
    private Map<String, String> typeData = new HashMap<String, String>();
    // 存储每个map用户分类记录数
    private Map<String, Integer> userType = new HashMap<String, Integer>();
    private String typeKey = null;

    // 预先加载
    @Override
    protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException{
        Path file = new Path(context.getCacheFiles()[0].getPath());
        BufferedReader br = new BufferedReader(new FileReader(file.toString()));
        String str = null;
        try {
            // 每一行读取
            while ((str = br.readLine()) != null) {
                String[] splits = str.split(TProperties.getValue("fileoutsplit"));
                // 行为ID，分类ID
                typeData.put(splits[1], splits[0]);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            br.close();
        }
    }

    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String values[] = value.toString().split(TProperties.getValue("fileoutsplit"));
        // 行为ID，用户标识， PV数
        // 判断包含行为ID对应的分类ID
        if (typeData.containsKey(values[0])) {
            // key: 分类ID，用户ID
            typeKey = typeData.get(values[0]) + "," + values[1];
            if (userType.containsKey(typeKey)) {
                // 累加key次数
                userType.put(typeKey, userType.get(typeKey) + Integer.parseInt(values[2]));
            } else {
                // 加key次数
                userType.put(typeKey, Integer.parseInt(values[2]));
            }
        }
    }

    // map之后执行， 输出map累加数据结果
    @Override
    protected void cleanup(Context context) throws IOException,InterruptedException {
        for (String key : userType.keySet()) {
            // 分类ID
            okey = new Text(key.split(",", -1)[0]);
            ovalue = new Text (key.split(",", -1)[1] + TProperties.getValue("outfilesplit") + userType.get(key));
            // 分类ID  用户ID，访问次数
            context.write(okey, ovalue);
        }
    }
}